package com.zwan.oam_rtc.imps;

import com.zwan.oam_rtc.constants.Const;
import com.zwan.oam_rtc.utils.messages.MessageHead;
import com.zwan.oam_rtc.utils.messages.MsgStruct;
import lombok.extern.slf4j.Slf4j;

import java.io.InputStream;
import java.net.InetAddress;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.text.MessageFormat;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/**
 * 不判断moduleno和ipaddress的绑定关系
 * @author 叶建生
 *
 */
@Slf4j
public class ServerHandler implements Runnable {

	private Socket socket;
	private int moduleno = 0;
	private InetAddress addr ;
	private static final ScheduledExecutorService schedule = Executors.newSingleThreadScheduledExecutor();
	private ScheduledFuture<?> future = null;
	private long lastCheckTime = System.currentTimeMillis();
	
	public ServerHandler(Socket socket){
		this.socket = socket;
		addr = socket.getInetAddress();
	}
	
	public Socket getSocket(){
		return this.socket;
	}
	
	public InetAddress getAddress(){
		return addr;
	}
	
	public String getConnectionStr(){
		return (new StringBuffer()).append(socket.getInetAddress().getHostAddress()).append(":").append(socket.getPort()).toString();
	}
	
	private void reScheduleCheck(){
		reScheduleCheck(Const.TIME_IDLE);
	}
	
	// 重新计划检测任务
	private void reScheduleCheck(long waitTime){
		if (future != null)
			future.cancel(false);
		future = schedule.schedule(new CheckConnection(this), waitTime, TimeUnit.MILLISECONDS);
	}
	
	public void run() {
		if (this.socket == null || !socket.isConnected()) return;
		String ip = addr.getHostAddress();  //获得客户端ip
		try {
			log.info("Connect start from " + ip);
			reScheduleCheck(2000);
		//没有关闭io资源，虽然在close（）中关闭了socket，io就关了，但是得等垃圾回收，太慢了，最好手动关闭
			InputStream in = socket.getInputStream();   
			while(true){
				byte[] bs = new byte[Const.MSG_HEAD_LEN];
				in.read(bs);
				MessageHead head = new MessageHead();
				head.parse(bs, ByteOrder.LITTLE_ENDIAN);
				// 合法性检测
				if (!head.isValid()){
					log.info(MessageFormat.format(
							"Recive invalid message from {0} [module={1}], connection will be close", 
							addr.toString(), head.getSender().getModule()));
					close();
					break;
				}
				// 如果本模块没有注册，则先注册
				if (moduleno == 0){  
					ClientHandler ch = JavaImp.getInstance().getClientHandlers().get(head.getSender().getModule());
					if (ch != null){
						log.info(MessageFormat.format("Same module [{0}] is connected by {1}, Connection {2} will close.",
								head.getReceiver().getModule(), ch.getConnectionStr(),
								addr.toString()));
						close();
						break;
					}
					ServerHandler sh = JavaImp.getInstance().getServerHandlers().get(head.getSender().getModule()); 
					if (sh != null && sh != this)
					{
						log.info(MessageFormat.format("Same module [{0}] is connected by {1}, Connection {2} will close.",
								head.getReceiver().getModule(), sh.getAddress().toString(), 
								addr.toString()));
						close();
						break;
					}
					moduleno = head.getSender().getModule();
					JavaImp.getInstance().getServerHandlers().put(moduleno, this);
					log.info(MessageFormat.format("Module [{0}] from {1} has connected.", moduleno, addr.toString()));
				}
//				reScheduleCheck();
				if (head.isCheckMessage()){
					// 连接检测包，返回
//					ByteBuffer bf1 = MessageHead.getCheckMessageBuff(head.getReceiver().getModule(), 
//							head.getSender().getModule());
//					bf1.order(ByteOrder.LITTLE_ENDIAN);
//					bf1.rewind();
				    lastCheckTime = System.currentTimeMillis();
					if (head.getBodyLen() > 0)
					{
						byte[] bs1 = new byte[head.getBodyLen()];
						in.read(bs1);
					}
//					socket.getOutputStream().write(bf1.array());
//					reScheduleCheck();
				}
				else if (head.getReceiver().getModule() == 0){
					log.info("Client connection closed, moduleno: " + moduleno + ", ip: " + ip);
					close();
					break;
				}
				else if (head.getReceiver().getModule() != (short)JavaImp.getInstance().getModuleNo()) // 此包不是发给本模块的
				{
					log.info("Recived message module is " + head.getReceiver().getModule()
							+ " but not me");
//					 实现转发功能
					byte[] bs1 = new byte[head.getBodyLen()];
					in.read(bs1);
					JavaImp.getInstance().sendMessage(head, bs1);
					continue;
				}
				else
				{
//					reScheduleCheck();
					// 正式的数据包
					byte[] bs1 = new byte[head.getBodyLen()];
					in.read(bs1);
					JavaImp.getInstance().getExecutor().submit(new HandlerRunner(head, ByteBuffer.wrap(bs1), JavaImp.getInstance().getByteOrder()));
				}
			}
		} catch (Exception e) {
			if (moduleno > 0)
			{
				log.info("Client connection closed, moduleno: " + moduleno + ", ip: " + ip);
			}
			else
				log.info("Client connection closed, ip: " + ip);
			close();
		}
	}

	public void close() {
		if (moduleno > 0)
			JavaImp.getInstance().getServerHandlers().remove(moduleno);
		if (socket != null && socket.isConnected()){
			try {
				socket.close();
			} catch (Exception e) {
//				log.error(e);
			}
		}
	}
	
	public int getModule(){
		return this.moduleno;
	}
	
	public void sendMessage(int srcpno, int descpno, int eventid, MsgStruct body){
		sendMessage(srcpno, descpno, eventid, body.getByteArray(JavaImp.getInstance().getByteOrder()));
	}
	
	public void sendMessage(int srcpno, int descpno, int eventid, byte[] body){
		MessageHead head = new MessageHead();
		head.getSender().setModule((short)JavaImp.getInstance().getModuleNo());
		head.getSender().setPno(srcpno);
		head.getReceiver().setModule((short)moduleno);
		head.getReceiver().setPno(descpno);
		head.setBodyLen(body.length);
		head.setSecretfield(head.calcSecretField());
		sendMessage(head, body);
	}
	
	public void sendMessage(MessageHead head, byte[] body){
		ByteBuffer bf = ByteBuffer.allocate(head.getLength() + body.length);
		head.writeToBuffer(bf);
		bf.put(body);
		bf.rewind();
		try {
			socket.getOutputStream().write(bf.array());
		} catch (Exception e) {
			log.info(MessageFormat.format("Send message to module [{0}] error, eventid: {1,number,#}, length: {2}",
					moduleno, head.getEventid(), body.length));
			close();
		}
	}
	
	class CheckConnection implements Runnable{
		private ServerHandler h;
		
		public CheckConnection(ServerHandler h){
			this.h = h;
		}
		
		public void run(){
		    if (System.currentTimeMillis() - h.lastCheckTime > Const.TIME_IDLE * 2){
		        h.future.cancel(true);
		        h.close();
		        this.h = null;
		        return;
		    }
			byte[] bs = MessageHead.getCheckMessage((byte)JavaImp.getInstance().getModuleNo(), 
					(byte)(moduleno==0?255:moduleno));
			try {
				h.socket.getOutputStream().write(bs);
				h.reScheduleCheck();
			} catch (Exception e) {
				h.future.cancel(true);
			}
		}
	}

}
